package com.ds.lens.flink.monitor.eventsinkjob.function;

import com.ds.lens.data.common.document.Event;
import com.ds.lens.flink.monitor.MonitorJobContext;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;
import org.elasticsearch.common.xcontent.XContentType;

import java.nio.charset.StandardCharsets;

public class MonitorEventSinkFunction implements ElasticsearchSinkFunction<Event> {

    public IndexRequest createIndexRequest(Event element) {
        String str = MonitorJobContext.getInstance().getGson().toJson(element);
        String eventIndex = MonitorJobContext.getInstance().getEsMonitorEventIndex();
        String eventType = MonitorJobContext.getInstance().getEsMonitorEventType();
        return Requests.indexRequest()
                .index(eventIndex)
                .type(eventType)
                .source(str.getBytes(StandardCharsets.UTF_8), XContentType.JSON);
    }

    @Override
    public void process(Event element, RuntimeContext ctx, RequestIndexer indexer) {
        indexer.add(createIndexRequest(element));
    }


}
